package com.webchat.ugc.service;

import com.google.common.collect.Lists;
import com.webchat.common.enums.MessageQueueTaskStatusEnum;
import com.webchat.common.service.messagequeue.task.AbstractMQTaskService;
import com.webchat.common.service.messagequeue.task.MessageQueueTaskDTO;
import com.webchat.ugc.repository.dao.IMessageQueueTaskDAO;
import com.webchat.ugc.repository.entity.MessageQueueTask;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;


@Service
public class UgcMessageQueueTaskService extends AbstractMQTaskService<MessageQueueTask> {

    @Autowired
    private IMessageQueueTaskDAO messageQueueTaskDAO;


    @Override
    protected List<MessageQueueTaskDTO> tasks() {
        List<MessageQueueTaskDTO> tasks = new ArrayList<>();
        Pageable pageable = PageRequest.of(0, QUERY_MAX_SIZE);
        Page<MessageQueueTask> page = messageQueueTaskDAO.findAllByStatus(MessageQueueTaskStatusEnum.ERROR.getStatus(), pageable);
        List<MessageQueueTask> taskEntities = page.getContent();
        if (CollectionUtils.isNotEmpty(taskEntities)) {
            // 直接取第一页
            tasks.addAll(super.convert(taskEntities));
        }
        if (page.getTotalPages() > 1) {
            // 超过一页，从第2页开始获取
            for (int i = 1; i < page.getTotalPages(); i++) {
                pageable = PageRequest.of(i, QUERY_MAX_SIZE);
                Page<MessageQueueTask> currPage = messageQueueTaskDAO.findAllByStatus(MessageQueueTaskStatusEnum.ERROR.getStatus(), pageable);
                List<MessageQueueTask> currTaskEntities = currPage.getContent();
                if (CollectionUtils.isNotEmpty(currTaskEntities)) {
                    tasks.addAll(super.convert(currTaskEntities));
                }
            }
        }
        return tasks;
    }

    @Override
    public int saveFailTask2DB(List<MessageQueueTaskDTO> tasks) {

        if (CollectionUtils.isEmpty(tasks)) {
            return 0;
        }

        List<MessageQueueTask> taskEntities = tasks.stream().map(task -> {
            MessageQueueTask messageQueueTask = new MessageQueueTask();
            BeanUtils.copyProperties(task, messageQueueTask);
            return messageQueueTask;
        }).toList();
        return messageQueueTaskDAO.saveAll(taskEntities).size();
    }

    @Override
    protected void updateTaskStatus(List<String> taskIds, MessageQueueTaskStatusEnum status) {

        if (CollectionUtils.isEmpty(taskIds)) {
            return;
        }
        List<List<String>> taskPartition = Lists.partition(taskIds, QUERY_MAX_SIZE);

        for (List<String> tasks : taskPartition) {
            messageQueueTaskDAO.batchUpdateStatus(tasks, status.getStatus());
        }
    }
}
